package com.neusoft.databus.sap.service.impl;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
import com.google.common.collect.Maps;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.common.utils.CacheUtil;
import com.neusoft.bizcore.web.support.Searchable;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.constant.DatabusConstant;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.dto.MonitorDataDTO;
import com.neusoft.databus.common.model.MonitorData;
import com.neusoft.databus.core.converter.MonitorDataConverter;
import com.neusoft.databus.core.metric.TpsEndpoint;
import com.neusoft.databus.core.repository.MonitorDataRepository;
import com.neusoft.databus.core.service.MonitorDataService;

import lombok.extern.slf4j.Slf4j;

/**
 * 监控数据服务 <br>
 * 异步接收消息总线发送的消息 {#processMonitorDataEvent}
 *
 * @author sunchf
 */
@Slf4j
@Service
public class MonitorDataServiceImpl implements MonitorDataService {

    @Autowired
    private MonitorDataRepository monitorDataRepository;
    @Autowired
    private MonitorDataConverter monitorDataConverter;
    @javax.annotation.Resource(name = "monitorDataEventBus")
    private AsyncEventBus monitorDataEventBus;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private TpsEndpoint tpsEndpoint;

    @PostConstruct
    public void init() {
        this.monitorDataEventBus.register(this);
    }

    @Override
    public List<MonitorDataDTO> search(final String resourceCode, final Searchable searchable) {

        int latestLimit = 10;
        if (searchable.hasKey("latestLimit")) {
            latestLimit = searchable.getIntValue("latestLimit");
        }

        List<MonitorData> list;
        switch (latestLimit) {
        case 20:
            list = this.monitorDataRepository.findFirst20ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        case 50:
            list = this.monitorDataRepository.findFirst50ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        case 100:
            list = this.monitorDataRepository.findFirst100ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        case 200:
            list = this.monitorDataRepository.findFirst200ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        case 500:
            list = this.monitorDataRepository.findFirst500ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        case 1000:
            list = this.monitorDataRepository.findFirst1000ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        default:
            list = this.monitorDataRepository.findFirst10ByResourceCodeOrderByCollectedTimeDesc(resourceCode);
            break;
        }

        final List<MonitorDataDTO> results = this.monitorDataConverter.toListDTO(list);
        final String[] qualifiers = searchable.getStrArray("qualifier");
        if (null != qualifiers) {
            final Set<String> qualifierSet = Arrays.asList(qualifiers).stream().collect(Collectors.toSet());
            results.stream().forEach(result -> {
                final Map<String, Object> map = JsonUtils.jsonToMap(result.getData());
                final String json = JsonUtils.pojoToJson(map,
                        SimpleBeanPropertyFilter.filterOutAllExcept(qualifierSet));

                result.setData(json);
            });

        }

        return this.monitorDataConverter.toListDTO(list);
    }

    /**
     * Event Bus事件处理
     *
     * @param commandBean
     * @return
     */
    @AllowConcurrentEvents // 线程安全
    @Subscribe
    public void processMonitorDataEvent(final CommandBean commandBean) {
        try {
            MDC.put("trace", commandBean.generateTraceId());
            MonitorDataServiceImpl.log.info("monitor-data-eventbus: ready to process monitor data: {}",
                    commandBean.generateTraceId());
            if (commandBean.isError() && !DatabusErrorCode.ERR_CODE_406005.equals(commandBean.getErrorCode())) {
                MonitorDataServiceImpl.log.info("monitor-data-eventbus: invalid monitor data, discard it.");
                return;
            }
            if (MonitorDataServiceImpl.log.isDebugEnabled()) {
                MonitorDataServiceImpl.log.debug("monitor-data-eventbus: invoke tps endpoint metric collector, {}",
                        this.tpsEndpoint);
            }
            this.tpsEndpoint.saveLatestTps(commandBean);
            this.saveCache(commandBean);
            MDC.clear();
        } catch (final Exception e) {
            MonitorDataServiceImpl.log.error(e.getMessage(), e);
        }
    }

    private void saveCache(final CommandBean commandBean) {

        commandBean.getResourceBean().forEach(r -> {
            final List<Map<String, Object>> metricData = r.getMetrics().stream().filter(i -> i.isParseSuccess())
                    .map(it -> {
                        final Map<String, Object> map = new HashMap<>();
                        map.put("name", it.getName());
                        map.put("code", it.getVariable());
                        map.put("value", it.getValue());
                        if (null != it.getUnit()) {
                            map.put("unit", it.getUnit());
                        }
                        return map;
                    }).collect(Collectors.toList());
            final Map<String, Object> monitor = Maps.newHashMap();
            monitor.put("collectedTime", commandBean.getCollectEndTime());
            monitor.put("elapsedTime", commandBean.getEndTime().getTime() - commandBean.getBeginTime().getTime());
            monitor.put("data", metricData);

            if (!metricData.isEmpty() && (metricData.size() > 0)) {
                this.cacheService.put(DatabusConstant.CACHE_NAME_MONITOR_DATA, this.getMonitorDataKey(r.getCode()),
                        monitor);
            }
        });

    }

    @SuppressWarnings("unchecked")
    @Override
    public Map<String, Object> findInCache(final String resourceCode) {
        Map<String, Object> data = Maps.newHashMap();
        final String key = this.getMonitorDataKey(resourceCode);
        if (this.cacheService.isInCache(DatabusConstant.CACHE_NAME_MONITOR_DATA, key)) {
            data = (Map<String, Object>) this.cacheService.get(DatabusConstant.CACHE_NAME_MONITOR_DATA, key);
        }
        data.put("code", resourceCode);
        return data;
    }

    private String getMonitorDataKey(final String resourceCode) {
        final StringBuffer key = new StringBuffer();
        key.append(CacheUtil.keyBuilder(resourceCode));
        return key.toString();
    }

    @SuppressWarnings("serial")
    @Override
    public Map<String, List<Map<String, Object>>> search(final List<String> resourceCodes, final long startTime,
            final long endTime) {
        final Date start = new Date(startTime);
        final Date end = new Date(endTime);
        final List<MonitorData> latest = this.monitorDataRepository.searchLatestData(resourceCodes, start, end);
        final List<MonitorData> oldest = this.monitorDataRepository.searchOldestData(resourceCodes, start, end);
        final Map<String, List<Map<String, Object>>> results = new HashMap<>();

        Arrays.asList(oldest, latest).forEach(arr -> {
            arr.stream().forEach(it -> {
                if (!results.containsKey(it.getResource().getCode())) {
                    results.put(it.getResource().getCode(), new ArrayList<>());
                }
                results.get(it.getResource().getCode()).add(new HashMap<String, Object>() {
                    {
                        this.put("resourceCode", it.getResource().getCode());
                        this.put("collectedTime", it.getCollectedTime().getTime());
                        this.put("data", it.getData());
                    }
                });
            });
        });

        return results;
    }

    @Override
    public Page<MonitorDataDTO> search(final Pageable pageable, final Searchable searchable) {
        final Page<MonitorData> models = this.monitorDataRepository.search(pageable, searchable);
        return this.monitorDataConverter.toPageDTO(models);
    }

    @Override
    public long countOfDay(final String resourceCode, final int day) {
        final Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.HOUR_OF_DAY, 0);
        calendar.set(Calendar.MINUTE, 0);
        calendar.set(Calendar.SECOND, 0);
        calendar.add(Calendar.DAY_OF_MONTH, day);

        final Date thatDay = calendar.getTime();
        calendar.add(Calendar.SECOND, 86399);
        final Date thatDayEnd = calendar.getTime();
        final long count = this.monitorDataRepository.countByResource_codeAndCreateTimeBetween(resourceCode, thatDay,
                thatDayEnd);
        return count;
    }

    @SuppressWarnings("unchecked")
    public List<Map<String, Object>> findInCache(final List<String> resourceCodeList) {

        final List<Map<String, Object>> result = new ArrayList<>();
        resourceCodeList.stream().forEach(code -> {
            Map<String, Object> data = Maps.newHashMap();
            final String key = this.getMonitorDataKey(code);
            if (this.cacheService.isInCache(DatabusConstant.CACHE_NAME_MONITOR_DATA, key)) {
                data = (Map<String, Object>) this.cacheService.get(DatabusConstant.CACHE_NAME_MONITOR_DATA, key);
            }
            data.put("code", code);
            result.add(data);
        });

        return result;
    }

}
